package org.wu.framework.easy.pulsar.listener;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.context.SmartLifecycle;
import org.wu.framework.core.exception.AssertFactory;
import org.wu.framework.easy.listener.core.ConcurrentMessageListenerContainer;
import org.wu.framework.easy.listener.core.GenericMessageListenerContainer;
import org.wu.framework.easy.listener.core.SingletonMessageListenerContainer;
import org.wu.framework.easy.listener.core.config.SingletonListenerContainerFactory;
import org.wu.framework.easy.pulsar.config.MethodPulsarListenerEndpoint;

import java.util.ArrayList;
import java.util.List;


public class PulsarConcurrentMessageListenerContainer<K, V> implements ConcurrentMessageListenerContainer<K, V>, GenericMessageListenerContainer<K, V> {

    protected List<PulsarSingletonMessageListenerContainer> containerList = new ArrayList<>();
    private int concurrency = 0;
    private PulsarClient pulsarClient;
    private MethodPulsarListenerEndpoint endpoint;

    public MethodPulsarListenerEndpoint getEndpoint() {
        return endpoint;
    }

    public void setEndpoint(MethodPulsarListenerEndpoint endpoint) {
        this.endpoint = endpoint;
    }

    public PulsarClient getPulsarClient() {
        return pulsarClient;
    }

    public void setPulsarClient(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
    }

    @Override
    public int getConcurrency() {
        return concurrency;
    }

    @Override
    public void setConcurrency(int concurrency) {
        this.concurrency = concurrency;
    }

    @Override
    public void setSingletonListenerContainerFactory(SingletonListenerContainerFactory containerFactory) {

    }

    @Override
    public SingletonMessageListenerContainer createContainer(int index) {
        final PulsarSingletonMessageListenerContainer listenerContainer = new PulsarSingletonMessageListenerContainer();
        listenerContainer.setRunning(true);
        AssertFactory.notNull(endpoint, "'endpoint' cannot be null");

        Consumer<byte[]> subscribe = null;
        try {
            // add topic
            subscribe = getPulsarClient()
                    .newConsumer()
                    .consumerName(endpoint.getSubscriptionName() + "-" + index + "C")
                    .topic(endpoint.getTopics().toArray(new String[0]))
                    .subscriptionName(endpoint.getSubscriptionName())
                    .subscriptionType(endpoint.getSubscriptionType())
                    .subscribe();
        } catch (PulsarClientException e) {
            e.printStackTrace();
        }
        AssertFactory.notNull(subscribe, "'subscribe' cannot be null");
        listenerContainer.setConsumer(subscribe);
        listenerContainer.setBeanName(endpoint.getBeanName());
        listenerContainer.setEndpoint(endpoint);
        containerList.add(listenerContainer);
        return listenerContainer;
    }

    /**
     * Stop this component, typically in a synchronous fashion, such that the component is
     * fully stopped upon return of this method. Consider implementing {@link SmartLifecycle}
     * and its {@code stop(Runnable)} variant when asynchronous stop behavior is necessary.
     * <p>Note that this stop notification is not guaranteed to come before destruction:
     * On regular shutdown, {@code Lifecycle} beans will first receive a stop notification
     * before the general destruction callbacks are being propagated; however, on hot
     * refresh during a context's lifetime or on aborted refresh attempts, a given bean's
     * destroy method will be called without any consideration of stop signals upfront.
     * <p>Should not throw an exception if the component is not running (not started yet).
     * <p>In the case of a container, this will propagate the stop signal to all components
     * that apply.
     *
     * @see SmartLifecycle#stop(Runnable)
     * @see DisposableBean#destroy()
     */
    @Override
    public void stop() {
        containerList.forEach(PulsarSingletonMessageListenerContainer::stop);

    }

    /**
     * Check whether this component is currently running.
     * <p>In the case of a container, this will return {@code true} only if <i>all</i>
     * components that apply are currently running.
     *
     * @return whether the component is currently running
     */
    @Override
    public boolean isRunning() {
        return false;
    }

}
